Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/layers/spnego.py: 28%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# SPDX-License-Identifier: GPL-2.0-only
2# This file is part of Scapy
3# See https://scapy.net/ for more information
4# Copyright (C) Gabriel Potter <gabriel[]potter[]fr>
6"""
7SPNEGO
9Implements parts of:
11- GSSAPI SPNEGO: RFC4178 > RFC2478
12- GSSAPI SPNEGO NEGOEX: [MS-NEGOEX]
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"""
19import os
20import struct
21from uuid import UUID
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.consts import WINDOWS
44from scapy.fields import (
45 FieldListField,
46 LEIntEnumField,
47 LEIntField,
48 LELongEnumField,
49 LELongField,
50 LEShortField,
51 MultipleTypeField,
52 PacketField,
53 PacketListField,
54 StrField,
55 StrFixedLenField,
56 UUIDEnumField,
57 UUIDField,
58 XStrFixedLenField,
59 XStrLenField,
60)
61from scapy.error import log_runtime
62from scapy.packet import Packet, bind_layers
63from scapy.utils import (
64 valid_ip,
65 valid_ip6,
66)
68from scapy.layers.gssapi import (
69 _GSSAPI_OIDS,
70 _GSSAPI_SIGNATURE_OIDS,
71 GSS_C_FLAGS,
72 GSS_C_NO_CHANNEL_BINDINGS,
73 GSS_S_BAD_MECH,
74 GSS_S_BAD_MIC,
75 GSS_S_COMPLETE,
76 GSS_S_CONTINUE_NEEDED,
77 GSS_S_FAILURE,
78 GSS_S_FLAGS,
79 GSSAPI_BLOB_SIGNATURE,
80 GSSAPI_BLOB,
81 GssChannelBindings,
82 SSP,
83)
85# SSP Providers
86from scapy.layers.kerberos import (
87 Kerberos,
88 KerberosSSP,
89 _parse_spn,
90 _parse_upn,
91)
92from scapy.layers.ntlm import (
93 NTLMSSP,
94 MD4le,
95 NEGOEX_EXCHANGE_NTLM,
96 NTLM_Header,
97 _NTLMPayloadField,
98 _NTLMPayloadPacket,
99)
101# Typing imports
102from typing import (
103 Dict,
104 List,
105 Optional,
106 Tuple,
107)
109# SPNEGO negTokenInit
110# https://datatracker.ietf.org/doc/html/rfc4178#section-4.2.1
113class SPNEGO_MechType(ASN1_Packet):
114 ASN1_codec = ASN1_Codecs.BER
115 ASN1_root = ASN1F_OID("oid", None)
118class SPNEGO_MechTypes(ASN1_Packet):
119 ASN1_codec = ASN1_Codecs.BER
120 ASN1_root = ASN1F_SEQUENCE_OF("mechTypes", None, SPNEGO_MechType)
123class SPNEGO_MechListMIC(ASN1_Packet):
124 ASN1_codec = ASN1_Codecs.BER
125 ASN1_root = ASN1F_STRING_ENCAPS("value", "", GSSAPI_BLOB_SIGNATURE)
128_mechDissector = {
129 "1.3.6.1.4.1.311.2.2.10": NTLM_Header, # NTLM
130 "1.2.840.48018.1.2.2": Kerberos, # MS KRB5 - Microsoft Kerberos 5
131 "1.2.840.113554.1.2.2": Kerberos, # Kerberos 5
132 "1.2.840.113554.1.2.2.3": Kerberos, # Kerberos 5 - User to User
133 "1.3.6.1.5.2.5": Kerberos, # Kerberos 5 - IAKERB
134}
137class _SPNEGO_Token_Field(ASN1F_STRING):
138 def i2m(self, pkt, x):
139 if x is None:
140 x = b""
141 return super(_SPNEGO_Token_Field, self).i2m(pkt, bytes(x))
143 def m2i(self, pkt, s):
144 dat, r = super(_SPNEGO_Token_Field, self).m2i(pkt, s)
145 types = None
146 if isinstance(pkt.underlayer, SPNEGO_negTokenInit):
147 types = pkt.underlayer.mechTypes
148 elif isinstance(pkt.underlayer, SPNEGO_negTokenResp):
149 types = [pkt.underlayer.supportedMech]
150 if types and types[0] and types[0].oid.val in _mechDissector:
151 return _mechDissector[types[0].oid.val](dat.val), r
152 else:
153 # Use heuristics
154 return GSSAPI_BLOB(dat.val), r
157class SPNEGO_Token(ASN1_Packet):
158 ASN1_codec = ASN1_Codecs.BER
159 ASN1_root = _SPNEGO_Token_Field("value", None)
162_ContextFlags = [
163 "delegFlag",
164 "mutualFlag",
165 "replayFlag",
166 "sequenceFlag",
167 "superseded",
168 "anonFlag",
169 "confFlag",
170 "integFlag",
171]
174class SPNEGO_negHints(ASN1_Packet):
175 # [MS-SPNG] 2.2.1
176 ASN1_codec = ASN1_Codecs.BER
177 ASN1_root = ASN1F_SEQUENCE(
178 ASN1F_optional(
179 ASN1F_GENERAL_STRING(
180 "hintName", "not_defined_in_RFC4178@please_ignore", explicit_tag=0xA0
181 ),
182 ),
183 ASN1F_optional(
184 ASN1F_GENERAL_STRING("hintAddress", None, explicit_tag=0xA1),
185 ),
186 )
189class SPNEGO_negTokenInit(ASN1_Packet):
190 ASN1_codec = ASN1_Codecs.BER
191 ASN1_root = ASN1F_SEQUENCE(
192 ASN1F_optional(
193 ASN1F_SEQUENCE_OF("mechTypes", None, SPNEGO_MechType, explicit_tag=0xA0)
194 ),
195 ASN1F_optional(ASN1F_FLAGS("reqFlags", None, _ContextFlags, implicit_tag=0x81)),
196 ASN1F_optional(
197 ASN1F_PACKET("mechToken", None, SPNEGO_Token, explicit_tag=0xA2),
198 ),
199 # [MS-SPNG] flavor !
200 ASN1F_optional(
201 ASN1F_PACKET("negHints", None, SPNEGO_negHints, explicit_tag=0xA3)
202 ),
203 ASN1F_optional(
204 ASN1F_PACKET("mechListMIC", None, SPNEGO_MechListMIC, explicit_tag=0xA4)
205 ),
206 # Compat with RFC 4178's SPNEGO_negTokenInit
207 ASN1F_optional(
208 ASN1F_PACKET("_mechListMIC", None, SPNEGO_MechListMIC, explicit_tag=0xA3)
209 ),
210 )
213# SPNEGO negTokenTarg
214# https://datatracker.ietf.org/doc/html/rfc4178#section-4.2.2
217class SPNEGO_negTokenResp(ASN1_Packet):
218 ASN1_codec = ASN1_Codecs.BER
219 ASN1_root = ASN1F_SEQUENCE(
220 ASN1F_optional(
221 ASN1F_ENUMERATED(
222 "negState",
223 0,
224 {
225 0: "accept-completed",
226 1: "accept-incomplete",
227 2: "reject",
228 3: "request-mic",
229 },
230 explicit_tag=0xA0,
231 ),
232 ),
233 ASN1F_optional(
234 ASN1F_PACKET(
235 "supportedMech", SPNEGO_MechType(), SPNEGO_MechType, explicit_tag=0xA1
236 ),
237 ),
238 ASN1F_optional(
239 ASN1F_PACKET("responseToken", None, SPNEGO_Token, explicit_tag=0xA2)
240 ),
241 ASN1F_optional(
242 ASN1F_PACKET("mechListMIC", None, SPNEGO_MechListMIC, explicit_tag=0xA3)
243 ),
244 # [MS-SPNG] Late Fallback Mechanism
245 ASN1F_optional(
246 ASN1F_PACKET("supportedMechs", None, SPNEGO_MechTypes, explicit_tag=0xA4)
247 ),
248 )
251class SPNEGO_negToken(ASN1_Packet):
252 ASN1_codec = ASN1_Codecs.BER
253 ASN1_root = ASN1F_CHOICE(
254 "token",
255 SPNEGO_negTokenInit(),
256 ASN1F_PACKET(
257 "negTokenInit",
258 SPNEGO_negTokenInit(),
259 SPNEGO_negTokenInit,
260 explicit_tag=0xA0,
261 ),
262 ASN1F_PACKET(
263 "negTokenResp",
264 SPNEGO_negTokenResp(),
265 SPNEGO_negTokenResp,
266 explicit_tag=0xA1,
267 ),
268 )
271# Register for the GSS API Blob
273_GSSAPI_OIDS["1.3.6.1.5.5.2"] = SPNEGO_negToken
274_GSSAPI_SIGNATURE_OIDS["1.3.6.1.5.5.2"] = SPNEGO_negToken
277def mechListMIC(oids):
278 """
279 Implementation of RFC 4178 - Appendix D. mechListMIC Computation
281 NOTE: The documentation on mechListMIC isn't super clear, so note that:
283 - The mechListMIC that the client sends is computed over the
284 list of mechanisms that it requests.
285 - the mechListMIC that the server sends is computed over the
286 list of mechanisms that the client requested.
288 This also means that NegTokenInit2 added by [MS-SPNG] is NOT protected.
289 That's not necessarily an issue, since it was optional in most cases,
290 but it's something to keep in mind.
291 """
292 return bytes(SPNEGO_MechTypes(mechTypes=oids))
295# NEGOEX
296# https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-negoex/0ad7a003-ab56-4839-a204-b555ca6759a2
299_NEGOEX_AUTH_SCHEMES = {
300 # Reversed. Is there any doc related to this?
301 # The NEGOEX doc is very ellusive
302 UUID("5c33530d-eaf9-0d4d-b2ec-4ae3786ec308"): "UUID('[NTLM-UUID]')",
303}
306class NEGOEX_MESSAGE_HEADER(Packet):
307 fields_desc = [
308 StrFixedLenField("Signature", "NEGOEXTS", length=8),
309 LEIntEnumField(
310 "MessageType",
311 0,
312 {
313 0x0: "INITIATOR_NEGO",
314 0x01: "ACCEPTOR_NEGO",
315 0x02: "INITIATOR_META_DATA",
316 0x03: "ACCEPTOR_META_DATA",
317 0x04: "CHALLENGE",
318 0x05: "AP_REQUEST",
319 0x06: "VERIFY",
320 0x07: "ALERT",
321 },
322 ),
323 LEIntField("SequenceNum", 0),
324 LEIntField("cbHeaderLength", None),
325 LEIntField("cbMessageLength", None),
326 UUIDField("ConversationId", None),
327 ]
329 def post_build(self, pkt, pay):
330 if self.cbHeaderLength is None:
331 pkt = pkt[16:] + struct.pack("<I", len(pkt)) + pkt[20:]
332 if self.cbMessageLength is None:
333 pkt = pkt[20:] + struct.pack("<I", len(pkt) + len(pay)) + pkt[24:]
334 return pkt + pay
337def _NEGOEX_post_build(self, p, pay_offset, fields):
338 # type: (Packet, bytes, int, Dict[str, Tuple[str, int]]) -> bytes
339 """Util function to build the offset and populate the lengths"""
340 for field_name, value in self.fields["Payload"]:
341 length = self.get_field("Payload").fields_map[field_name].i2len(self, value)
342 count = self.get_field("Payload").fields_map[field_name].i2count(self, value)
343 offset = fields[field_name]
344 # Offset
345 if self.getfieldval(field_name + "BufferOffset") is None:
346 p = p[:offset] + struct.pack("<I", pay_offset) + p[offset + 4 :]
347 # Count
348 if self.getfieldval(field_name + "Count") is None:
349 p = p[: offset + 4] + struct.pack("<H", count) + p[offset + 6 :]
350 pay_offset += length
351 return p
354class NEGOEX_BYTE_VECTOR(Packet):
355 fields_desc = [
356 LEIntField("ByteArrayBufferOffset", 0),
357 LEIntField("ByteArrayLength", 0),
358 ]
360 def guess_payload_class(self, payload):
361 return conf.padding_layer
364class NEGOEX_EXTENSION_VECTOR(Packet):
365 fields_desc = [
366 LELongField("ExtensionArrayOffset", 0),
367 LEShortField("ExtensionCount", 0),
368 ]
371class NEGOEX_NEGO_MESSAGE(_NTLMPayloadPacket):
372 OFFSET = 92
373 show_indent = 0
374 fields_desc = [
375 NEGOEX_MESSAGE_HEADER,
376 XStrFixedLenField("Random", b"", length=32),
377 LELongField("ProtocolVersion", 0),
378 LEIntField("AuthSchemeBufferOffset", None),
379 LEShortField("AuthSchemeCount", None),
380 LEIntField("ExtensionBufferOffset", None),
381 LEShortField("ExtensionCount", None),
382 # Payload
383 _NTLMPayloadField(
384 "Payload",
385 OFFSET,
386 [
387 FieldListField(
388 "AuthScheme",
389 [],
390 UUIDEnumField("", None, _NEGOEX_AUTH_SCHEMES),
391 count_from=lambda pkt: pkt.AuthSchemeCount,
392 ),
393 PacketListField(
394 "Extension",
395 [],
396 NEGOEX_EXTENSION_VECTOR,
397 count_from=lambda pkt: pkt.ExtensionCount,
398 ),
399 ],
400 length_from=lambda pkt: pkt.cbMessageLength - 92,
401 ),
402 # TODO: dissect extensions
403 ]
405 def post_build(self, pkt, pay):
406 # type: (bytes, bytes) -> bytes
407 return (
408 _NEGOEX_post_build(
409 self,
410 pkt,
411 self.OFFSET,
412 {
413 "AuthScheme": 96,
414 "Extension": 102,
415 },
416 )
417 + pay
418 )
420 @classmethod
421 def dispatch_hook(cls, _pkt=None, *args, **kargs):
422 if _pkt and len(_pkt) >= 12:
423 MessageType = struct.unpack("<I", _pkt[8:12])[0]
424 if MessageType in [0, 1]:
425 return NEGOEX_NEGO_MESSAGE
426 elif MessageType in [2, 3]:
427 return NEGOEX_EXCHANGE_MESSAGE
428 return cls
431# RFC3961
432_checksum_types = {
433 1: "CRC32",
434 2: "RSA-MD4",
435 3: "RSA-MD4-DES",
436 4: "DES-MAC",
437 5: "DES-MAC-K",
438 6: "RSA-MDA-DES-K",
439 7: "RSA-MD5",
440 8: "RSA-MD5-DES",
441 9: "RSA-MD5-DES3",
442 10: "SHA1",
443 12: "HMAC-SHA1-DES3-KD",
444 13: "HMAC-SHA1-DES3",
445 14: "SHA1",
446 15: "HMAC-SHA1-96-AES128",
447 16: "HMAC-SHA1-96-AES256",
448}
451def _checksum_size(pkt):
452 if pkt.ChecksumType == 1:
453 return 4
454 elif pkt.ChecksumType in [2, 4, 6, 7]:
455 return 16
456 elif pkt.ChecksumType in [3, 8, 9]:
457 return 24
458 elif pkt.ChecksumType == 5:
459 return 8
460 elif pkt.ChecksumType in [10, 12, 13, 14, 15, 16]:
461 return 20
462 return 0
465class NEGOEX_CHECKSUM(Packet):
466 fields_desc = [
467 LELongField("cbHeaderLength", 20),
468 LELongEnumField("ChecksumScheme", 1, {1: "CHECKSUM_SCHEME_RFC3961"}),
469 LELongEnumField("ChecksumType", None, _checksum_types),
470 XStrLenField("ChecksumValue", b"", length_from=_checksum_size),
471 ]
474class NEGOEX_EXCHANGE_MESSAGE(_NTLMPayloadPacket):
475 OFFSET = 64
476 show_indent = 0
477 fields_desc = [
478 NEGOEX_MESSAGE_HEADER,
479 UUIDEnumField("AuthScheme", None, _NEGOEX_AUTH_SCHEMES),
480 LEIntField("ExchangeBufferOffset", 0),
481 LEIntField("ExchangeLen", 0),
482 _NTLMPayloadField(
483 "Payload",
484 OFFSET,
485 [
486 # The NEGOEX doc mentions the following blob as as an
487 # "opaque handshake for the client authentication scheme".
488 # NEGOEX_EXCHANGE_NTLM is a reversed interpretation, and is
489 # probably not accurate.
490 MultipleTypeField(
491 [
492 (
493 PacketField("Exchange", None, NEGOEX_EXCHANGE_NTLM),
494 lambda pkt: pkt.AuthScheme
495 == UUID("5c33530d-eaf9-0d4d-b2ec-4ae3786ec308"),
496 ),
497 ],
498 StrField("Exchange", b""),
499 )
500 ],
501 length_from=lambda pkt: pkt.cbMessageLength - pkt.cbHeaderLength,
502 ),
503 ]
506class NEGOEX_VERIFY_MESSAGE(Packet):
507 show_indent = 0
508 fields_desc = [
509 NEGOEX_MESSAGE_HEADER,
510 UUIDEnumField("AuthScheme", None, _NEGOEX_AUTH_SCHEMES),
511 PacketField("Checksum", NEGOEX_CHECKSUM(), NEGOEX_CHECKSUM),
512 ]
515bind_layers(NEGOEX_NEGO_MESSAGE, NEGOEX_NEGO_MESSAGE)
518_mechDissector["1.3.6.1.4.1.311.2.2.30"] = NEGOEX_NEGO_MESSAGE
520# -- SSP
523class SPNEGOSSP(SSP):
524 """
525 The SPNEGO SSP
527 :param ssps: a dict with keys being the SSP class, and the value being a
528 dictionary of the keyword arguments to pass it on init.
530 Example::
532 from scapy.layers.ntlm import NTLMSSP
533 from scapy.layers.kerberos import KerberosSSP
534 from scapy.layers.spnego import SPNEGOSSP
535 from scapy.layers.smbserver import smbserver
536 from scapy.libs.rfc3961 import Encryption, Key
538 ssp = SPNEGOSSP([
539 NTLMSSP(
540 IDENTITIES={
541 "User1": MD4le("Password1"),
542 "Administrator": MD4le("Password123!"),
543 }
544 ),
545 KerberosSSP(
546 SPN="cifs/server2.domain.local",
547 KEY=Key(
548 Encryption.AES256,
549 key=hex_bytes("5e9255c907b2f7e969ddad816eabbec8f1f7a387c7194ecc98b827bdc9421c2b")
550 )
551 )
552 ])
553 smbserver(ssp=ssp)
554 """
556 __slots__ = [
557 "ssps",
558 "SUPPORT_LATE_FALLBACK",
559 ]
561 auth_type = 0x09
563 class STATE(SSP.STATE):
564 FIRST = 1
565 SUBSEQUENT = 2
566 MICONLY = 3
568 class CONTEXT(SSP.CONTEXT):
569 __slots__ = [
570 "req_flags",
571 "IsAcceptor",
572 "ssps",
573 "server_mechtypes",
574 "client_mechtypes",
575 "first_choice",
576 "require_mic",
577 "verified_mic",
578 "late_fallback_supported",
579 "late_fallback_negotiated",
580 "ssp",
581 "ssp_context",
582 "ssp_mechtype",
583 "raw",
584 ]
586 def __init__(
587 self,
588 IsAcceptor: bool,
589 ssps: List[SSP],
590 late_fallback_supported: bool,
591 req_flags=None,
592 ):
593 self.state = SPNEGOSSP.STATE.FIRST
594 self.req_flags = req_flags
595 self.IsAcceptor = IsAcceptor
596 # Information used during negotiation
597 self.ssps = ssps
598 self.server_mechtypes = None # the mechtypes the server requested
599 self.client_mechtypes = None # the mechtypes the client requested
600 self.first_choice = True # whether the SSP was the peer's first choice
601 self.require_mic = False # whether the mechListMIC is required or not
602 self.verified_mic = False # whether mechListMIC has been verified
603 self.late_fallback_supported = late_fallback_supported
604 self.late_fallback_negotiated = False # [MS-SPNG] exclusive
605 # Information about the currently selected SSP
606 self.ssp = None
607 self.ssp_context = None
608 self.ssp_mechtype = None
609 self.raw = False # fallback to raw SSP
610 super(SPNEGOSSP.CONTEXT, self).__init__()
612 # This is the order Windows chooses
613 _PREF_ORDER = [
614 "1.2.840.48018.1.2.2", # MS KRB5
615 "1.2.840.113554.1.2.2", # Kerberos 5
616 "1.3.6.1.5.2.5", # Kerberos 5 - IAKERB
617 "1.3.6.1.4.1.311.2.2.30", # NEGOEX
618 "1.2.840.113554.1.2.2.3", # Kerberos 5 - User to User
619 "1.3.6.1.4.1.311.2.2.10", # NTLM
620 "1.3.6.1.4.1.311.2.2.40", # Late Fallback
621 ]
623 def get_supported_mechtypes(self):
624 """
625 Return an ordered list of mechtypes that are still available.
626 """
627 # 1. Build mech list
628 mechs = []
629 for ssp in self.ssps:
630 mechs.extend(ssp.GSS_Inquire_names_for_mech())
632 # Windows 24H2 / Server 2025+
633 if self.late_fallback_supported:
634 mechs.append("1.3.6.1.4.1.311.2.2.40") # SPNEGO Late Fallback
636 # 2. Sort according to the selected SSP, then the preference order
637 selected_mech_oids = (
638 self.ssp.GSS_Inquire_names_for_mech() if self.ssp else []
639 )
640 mechs.sort(
641 key=lambda x: (x not in selected_mech_oids, self._PREF_ORDER.index(x))
642 )
644 # 4. Return wrapped in MechType
645 return [SPNEGO_MechType(oid=ASN1_OID(oid)) for oid in mechs]
647 def get_mechListMIC(self):
648 """
649 Return the binary used for mechListMIC
650 """
651 # See help(mechListMIC) for more details
652 mechtypes = self.client_mechtypes[:]
654 # [MS-SPNG] sect 3.1.5.1 - Late Fallback
655 # "When Negotiate Late Fallback is supported by both parties,
656 # mechListMIC consumes a list of all exchanged mechTypes and
657 # supportedMechs per the order of over-the-wire transmission
658 # with delimiters"
659 if self.late_fallback_negotiated:
660 # XXX Doesn't work. FIXME
661 mechtypes += self.server_mechtypes[:]
663 return mechListMIC(mechtypes)
665 def negotiate_ssp(self) -> None:
666 """
667 Perform SSP negotiation.
669 This updates our context and sets it with the first SSP that is
670 common to both client and server. This also applies rules from
671 [MS-SPNG] and RFC4178 to determine if mechListMIC is required.
672 """
673 if not self.IsAcceptor:
674 other_mechtypes = self.server_mechtypes
675 else:
676 other_mechtypes = self.client_mechtypes
678 if other_mechtypes is None:
679 # We don't have any information about the peer's preferred SSPs.
680 # This typically happens on client side, when NegTokenInit2 isn't used.
681 self.ssp = self.ssps[0]
682 ssp_oid = self.ssp.GSS_Inquire_names_for_mech()[0]
683 else:
684 # Get first common SSP between us and our peer
685 other_oids = [x.oid.val for x in other_mechtypes]
687 # See if the peer supports SPNEGO Late Fallback
688 if (
689 self.late_fallback_supported
690 and "1.3.6.1.4.1.311.2.2.40" in other_oids
691 ):
692 self.late_fallback_negotiated = True
694 # Find first common SSP
695 try:
696 self.ssp, ssp_oid = next(
697 (ssp, requested_oid)
698 for requested_oid in other_oids
699 for ssp in self.ssps
700 if requested_oid in ssp.GSS_Inquire_names_for_mech()
701 )
702 except StopIteration:
703 raise ValueError(
704 "Could not find a common SSP with the remote peer !"
705 )
707 # Check whether the selected SSP was the one preferred by the client
708 self.first_choice = ssp_oid == other_oids[0]
710 # Check whether mechListMIC is mandatory for this exchange
711 if not self.first_choice:
712 # RFC4178 rules for mechListMIC: mandatory if not the first choice.
713 self.require_mic = True
714 elif ssp_oid == "1.3.6.1.4.1.311.2.2.10" and self.ssp.SupportsMechListMIC():
715 # [MS-SPNG] note 8: "If NTLM authentication is most preferred by
716 # the client and the server, and the client includes a MIC in
717 # AUTHENTICATE_MESSAGE, then the mechListMIC field becomes
718 # mandatory"
719 self.require_mic = True
721 # Get the associated ssp dissection class and mechtype
722 self.ssp_mechtype = SPNEGO_MechType(oid=ASN1_OID(ssp_oid))
724 # Reset the ssp context
725 self.ssp_context = None
727 # Passthrough attributes and functions
729 def clifailure(self):
730 if self.ssp_context is not None:
731 self.ssp_context.clifailure()
733 def __getattr__(self, attr):
734 try:
735 return object.__getattribute__(self, attr)
736 except AttributeError:
737 return getattr(self.ssp_context, attr)
739 def __setattr__(self, attr, val):
740 try:
741 return object.__setattr__(self, attr, val)
742 except AttributeError:
743 return setattr(self.ssp_context, attr, val)
745 # Passthrough the flags property
747 @property
748 def flags(self):
749 if self.ssp_context:
750 return self.ssp_context.flags
751 return GSS_C_FLAGS(0)
753 @flags.setter
754 def flags(self, x):
755 if not self.ssp_context:
756 return
757 self.ssp_context.flags = x
759 def __repr__(self):
760 return "SPNEGOSSP[%s]" % repr(self.ssp_context)
762 def __init__(
763 self,
764 ssps: List[SSP],
765 SUPPORT_LATE_FALLBACK=False,
766 **kwargs,
767 ):
768 self.ssps = ssps
769 self.SUPPORT_LATE_FALLBACK = SUPPORT_LATE_FALLBACK
770 super(SPNEGOSSP, self).__init__(**kwargs)
772 @classmethod
773 def from_cli_arguments(
774 cls,
775 UPN: str,
776 target: str,
777 password: str = None,
778 HashNt: bytes = None,
779 HashAes256Sha96: bytes = None,
780 HashAes128Sha96: bytes = None,
781 kerberos_required: bool = False,
782 ST=None,
783 TGT=None,
784 KEY=None,
785 ccache: str = None,
786 debug: int = 0,
787 use_krb5ccname: bool = False,
788 use_winssp: bool = False,
789 ):
790 """
791 Initialize a SPNEGOSSP from a list of many arguments.
793 This is useful in a CLI, as it will try to build the best SPNEGOSSP
794 with NTLM and Kerberos based on the various parameters.
796 :param UPN: the UPN of the user to use.
797 :param target: the target IP/hostname entered by the user.
798 :param kerberos_required: require kerberos
799 :param password: (string) if provided, used for auth
800 :param HashNt: (bytes) if provided, used for auth (NTLM)
801 :param HashAes256Sha96: (bytes) if provided, used for auth (Kerberos)
802 :param HashAes128Sha96: (bytes) if provided, used for auth (Kerberos)
803 :param ST: if provided, the service ticket to use (Kerberos)
804 :param TGT: if provided, the TGT to use (Kerberos)
805 :param KEY: if ST provided, the session key associated to the ticket (Kerberos).
806 This can be either for the ST or TGT. Else, the user secret key.
807 :param ccache: (str) if provided, a path to a CCACHE (Kerberos)
808 :param use_krb5ccname: (bool) if true, the KRB5CCNAME environment variable will
809 be used if available.
810 :param use_winssp: (bool) (only works on Windows). Use implicit authentication
811 through WinSSP.
812 """
813 kerberos = True
814 domain_auth = True
815 hostname = None
816 # Check if target is a hostname / Check IP
817 if target and ":" in target:
818 if not valid_ip6(target):
819 hostname = target
820 else:
821 if not valid_ip(target):
822 hostname = target
824 # If using WinSSP, this goes fast.
825 if use_winssp:
826 if not WINDOWS:
827 raise OSError("Cannot use WinSSP on a non-Windows computer !")
828 from scapy.arch.windows.sspi import WinSSP
830 return WinSSP()
832 # Check UPN
833 try:
834 _, realm = _parse_upn(UPN)
835 if realm == ".":
836 # Local
837 domain_auth = False
838 except ValueError:
839 # not a UPN
840 if hostname is not None:
841 # Fallback to support IAKERB without a UPN
842 domain_auth = False
843 realm = hostname
844 UPN = f"{UPN}@{realm}"
845 else:
846 # NTLM only
847 kerberos = False
849 # If we're asked, check the environment for KRB5CCNAME
850 if use_krb5ccname and ccache is None and "KRB5CCNAME" in os.environ:
851 ccache = os.environ["KRB5CCNAME"]
853 # Do we need to ask the password?
854 if all(
855 x is None
856 for x in [
857 ST,
858 password,
859 HashNt,
860 HashAes256Sha96,
861 HashAes128Sha96,
862 ccache,
863 ]
864 ):
865 # yes.
866 from prompt_toolkit import prompt
868 password = prompt("Password: ", is_password=True)
870 ssps = []
871 # Kerberos
872 if kerberos and hostname:
873 # Get ticket if we don't already have one.
874 if ST is None and TGT is None and ccache is not None:
875 # In this case, load the KerberosSSP from ccache
876 from scapy.modules.ticketer import Ticketer
878 # Import into a Ticketer object
879 t = Ticketer()
880 t.open_ccache(ccache)
882 # Look for the ticket that we'll use. We chose:
883 # - either a ST if the UPN and SPN matches our target
884 # - or a ST that matches the UPN
885 # - else a TGT if we got nothing better
886 tgts = []
887 sts = []
888 for i, (tkt, key, upn, spn) in t.enumerate_tickets():
889 spn, _ = _parse_spn(spn)
890 spn_host = spn.split("/")[-1]
891 # Check that it's for the correct user
892 if upn.lower() == UPN.lower():
893 # Check that it's either a TGT or a ST to the correct service
894 if spn.lower().startswith("krbtgt/"):
895 # TGT. Keep it, and see if we don't have a better ST.
896 tgts.append(t.ssp(i))
897 elif hostname.lower() == spn_host.lower():
898 # ST. UPN and SPN match. We're done !
899 ssps.append(t.ssp(i))
900 break
901 else:
902 # ST. UPN matches, Keep it
903 sts.append(t.ssp(i))
904 else:
905 # No perfect ticket found
906 if tgts:
907 # Using a TGT !
908 ssps.append(tgts[0])
909 elif sts:
910 # Using a ST where at least the UPN matched !
911 ssps.append(sts[0])
912 else:
913 # Nothing found
914 t.show()
915 raise ValueError(
916 f"Could not find a ticket for {upn}, either a "
917 f"TGT or towards {hostname}"
918 )
919 elif ST is None and TGT is None:
920 # In this case, KEY is supposed to be the user's key.
921 from scapy.libs.rfc3961 import Key, EncryptionType
923 if KEY is None and HashAes256Sha96:
924 KEY = Key(
925 EncryptionType.AES256_CTS_HMAC_SHA1_96,
926 HashAes256Sha96,
927 )
928 elif KEY is None and HashAes128Sha96:
929 KEY = Key(
930 EncryptionType.AES128_CTS_HMAC_SHA1_96,
931 HashAes128Sha96,
932 )
933 elif KEY is None and HashNt:
934 KEY = Key(
935 EncryptionType.RC4_HMAC,
936 HashNt,
937 )
939 # We have a UPN and secret. This allows to support 3 cases:
940 # Kerberos, IAKerb and U2U
942 # Normal Kerberos and U2U are only supported in domain environments
943 if domain_auth:
944 ssps.extend(
945 [
946 KerberosSSP(
947 UPN=UPN,
948 PASSWORD=password,
949 KEY=KEY,
950 debug=debug,
951 ),
952 KerberosSSP(
953 UPN=UPN,
954 PASSWORD=password,
955 KEY=KEY,
956 debug=debug,
957 U2U=True,
958 ),
959 ]
960 )
962 # IAKERB is always supported when Kerberos is active
963 ssps.append(
964 KerberosSSP(
965 UPN=UPN,
966 PASSWORD=password,
967 KEY=KEY,
968 debug=debug,
969 IAKERB=True,
970 )
971 )
972 else:
973 # We have a ST, use it with the key.
974 ssps.append(
975 KerberosSSP(
976 UPN=UPN,
977 ST=ST,
978 TGT=TGT,
979 KEY=KEY,
980 debug=debug,
981 )
982 )
983 elif kerberos_required:
984 raise ValueError(
985 "Kerberos required but domain not specified in the UPN, "
986 "or target isn't a hostname !"
987 )
989 # NTLM
990 if not kerberos_required:
991 if HashNt is None and password is not None:
992 HashNt = MD4le(password)
993 if HashNt is not None:
994 ssps.append(NTLMSSP(UPN=UPN, HASHNT=HashNt))
996 if not ssps:
997 raise ValueError("Unexpected case ! Please report.")
999 # Build the SSP
1000 return cls(ssps)
1002 def NegTokenInit2(self):
1003 """
1004 Server-Initiation of GSSAPI/SPNEGO.
1005 See [MS-SPNG] sect 3.2.5.2
1006 """
1007 Context = SPNEGOSSP.CONTEXT(
1008 IsAcceptor=True,
1009 ssps=list(self.ssps),
1010 late_fallback_supported=self.SUPPORT_LATE_FALLBACK,
1011 )
1012 return (
1013 Context,
1014 GSSAPI_BLOB(
1015 innerToken=SPNEGO_negToken(
1016 token=SPNEGO_negTokenInit(
1017 mechTypes=Context.get_supported_mechtypes(),
1018 negHints=SPNEGO_negHints(
1019 hintName=ASN1_GENERAL_STRING(
1020 "not_defined_in_RFC4178@please_ignore"
1021 ),
1022 ),
1023 )
1024 )
1025 ),
1026 )
1028 # NOTE: NegoEX has an effect on how the SecurityContext is
1029 # initialized, as detailed in [MS-AUTHSOD] sect 3.3.2
1030 # But the format that the Exchange token uses appears not to
1031 # be documented :/
1033 # resp.SecurityBlob.innerToken.token.mechTypes.insert(
1034 # 0,
1035 # # NEGOEX
1036 # SPNEGO_MechType(oid="1.3.6.1.4.1.311.2.2.30"),
1037 # )
1038 # resp.SecurityBlob.innerToken.token.mechToken = SPNEGO_Token(
1039 # value=negoex_token
1040 # ) # noqa: E501
1042 def GSS_WrapEx(self, Context, *args, **kwargs):
1043 # Passthrough
1044 return Context.ssp.GSS_WrapEx(Context.ssp_context, *args, **kwargs)
1046 def GSS_UnwrapEx(self, Context, *args, **kwargs):
1047 # Passthrough
1048 return Context.ssp.GSS_UnwrapEx(Context.ssp_context, *args, **kwargs)
1050 def GSS_GetMICEx(self, Context, *args, **kwargs):
1051 # Passthrough
1052 return Context.ssp.GSS_GetMICEx(Context.ssp_context, *args, **kwargs)
1054 def GSS_VerifyMICEx(self, Context, *args, **kwargs):
1055 # Passthrough
1056 return Context.ssp.GSS_VerifyMICEx(Context.ssp_context, *args, **kwargs)
1058 def LegsAmount(self, Context: CONTEXT):
1059 return 4
1061 def MapStatusToNegState(self, status: int) -> int:
1062 """
1063 Map a GSSAPI return code to SPNEGO negState codes
1064 """
1065 if status == GSS_S_COMPLETE:
1066 return 0 # accept_completed
1067 elif status == GSS_S_CONTINUE_NEEDED:
1068 return 1 # accept_incomplete
1069 else:
1070 return 2 # reject
1072 def GuessOtherMechtypes(self, Context: CONTEXT, input_token):
1073 """
1074 Guesses the mechtype of the peer when the "raw" fallback is used.
1075 """
1076 if isinstance(input_token, NTLM_Header):
1077 other_mechtypes = [SPNEGO_MechType(oid=ASN1_OID("1.3.6.1.4.1.311.2.2.10"))]
1078 elif isinstance(input_token, Kerberos):
1079 other_mechtypes = [SPNEGO_MechType(oid=ASN1_OID("1.2.840.48018.1.2.2"))]
1080 else:
1081 other_mechtypes = []
1083 if not Context.IsAcceptor:
1084 Context.server_mechtypes = other_mechtypes
1085 else:
1086 Context.client_mechtypes = other_mechtypes
1088 def GSS_Init_sec_context(
1089 self,
1090 Context: CONTEXT,
1091 input_token=None,
1092 target_name: Optional[str] = None,
1093 req_flags: Optional[GSS_C_FLAGS] = None,
1094 chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,
1095 ):
1096 if Context is None:
1097 # New Context
1098 Context = SPNEGOSSP.CONTEXT(
1099 IsAcceptor=False,
1100 ssps=list(self.ssps),
1101 late_fallback_supported=self.SUPPORT_LATE_FALLBACK,
1102 req_flags=req_flags,
1103 )
1105 input_token_inner = None
1106 negState = None
1108 # Extract values from GSSAPI token, if present
1109 if input_token is not None:
1110 if isinstance(input_token, GSSAPI_BLOB):
1111 input_token = input_token.innerToken
1112 if isinstance(input_token, SPNEGO_negToken):
1113 input_token = input_token.token
1114 if isinstance(input_token, SPNEGO_negTokenInit):
1115 # We are handling a NegTokenInit2 request !
1116 # Populate context with values from the server's request
1117 Context.server_mechtypes = input_token.mechTypes
1118 elif isinstance(input_token, SPNEGO_negTokenResp):
1119 # Extract token and state from the client request
1120 if input_token.responseToken is not None:
1121 input_token_inner = input_token.responseToken.value
1122 if input_token.negState is not None:
1123 negState = input_token.negState
1125 # [MS-SPNG] Late Fallback Mechanism
1126 if input_token.supportedMechs is not None:
1127 Context.server_mechtypes = input_token.supportedMechs.mechTypes
1128 else:
1129 # The blob is a raw token. We aren't using SPNEGO here.
1130 Context.raw = True
1131 input_token_inner = input_token
1132 self.GuessOtherMechtypes(Context, input_token)
1134 # Perform SSP negotiation
1135 if Context.ssp is None:
1136 try:
1137 Context.negotiate_ssp()
1138 except ValueError as ex:
1139 # Couldn't find common SSP
1140 log_runtime.warning("SPNEGOSSP: %s" % ex)
1141 return Context, None, GSS_S_BAD_MECH
1143 if Context.state == SPNEGOSSP.STATE.MICONLY:
1144 # We have already finished the inner-ssp, and are just doing
1145 # an extra exchange because we were asked for the mechListMIC.
1146 if negState == 0:
1147 return Context, None, GSS_S_COMPLETE
1148 else:
1149 return Context, None, GSS_S_BAD_MIC
1151 # Call inner-SSP
1152 Context.ssp_context, output_token_inner, status = (
1153 Context.ssp.GSS_Init_sec_context(
1154 Context.ssp_context,
1155 input_token=input_token_inner,
1156 target_name=target_name,
1157 req_flags=Context.req_flags,
1158 chan_bindings=chan_bindings,
1159 )
1160 )
1162 if negState == 2 or status not in [GSS_S_COMPLETE, GSS_S_CONTINUE_NEEDED]:
1163 # SSP failed. Remove it from the list of SSPs we're currently running
1164 Context.ssps.remove(Context.ssp)
1165 log_runtime.warning(
1166 "SPNEGOSSP: %s failed. Retrying with next in queue." % repr(Context.ssp)
1167 )
1169 if Context.ssps:
1170 # We have other SSPs remaining. Retry using another one.
1171 Context.ssp = None
1172 return self.GSS_Init_sec_context(
1173 Context,
1174 None, # No input for retry.
1175 target_name=target_name,
1176 req_flags=req_flags,
1177 chan_bindings=chan_bindings,
1178 )
1179 else:
1180 # We don't have anything left
1181 return Context, None, status
1183 # Raw processing ends here.
1184 if Context.raw:
1185 return Context, output_token_inner, status
1187 # Verify MIC if present.
1188 if status == GSS_S_COMPLETE and input_token and input_token.mechListMIC:
1189 # NOTE: the mechListMIC that the server sends is computed over the list of
1190 # mechanisms that the **client requested**.
1191 Context.ssp.VerifyMechListMIC(
1192 Context.ssp_context,
1193 input_token.mechListMIC.value,
1194 Context.get_mechListMIC(),
1195 )
1196 Context.verified_mic = True
1198 if negState == 0 and status == GSS_S_COMPLETE:
1199 # We are done.
1200 return Context, None, status
1201 elif Context.state == SPNEGOSSP.STATE.FIRST:
1202 # First freeze the list of available mechtypes on the first message
1203 Context.client_mechtypes = Context.get_supported_mechtypes()
1205 # Now build the token
1206 spnego_tok = GSSAPI_BLOB(
1207 innerToken=SPNEGO_negToken(
1208 token=SPNEGO_negTokenInit(mechTypes=Context.client_mechtypes)
1209 )
1210 )
1212 # Add the output token if provided
1213 if output_token_inner is not None:
1214 spnego_tok.innerToken.token.mechToken = SPNEGO_Token(
1215 value=output_token_inner,
1216 )
1217 elif Context.state == SPNEGOSSP.STATE.SUBSEQUENT:
1218 # Build subsequent client tokens: without the list of supported mechtypes
1219 # NOTE: GSSAPI_BLOB is stripped.
1220 spnego_tok = SPNEGO_negToken(
1221 token=SPNEGO_negTokenResp(
1222 supportedMech=None,
1223 negState=None,
1224 )
1225 )
1227 # Add the MIC if required and the exchange is finished.
1228 if status == GSS_S_COMPLETE and Context.require_mic:
1229 spnego_tok.token.mechListMIC = SPNEGO_MechListMIC(
1230 value=Context.ssp.GetMechListMIC(
1231 Context.ssp_context,
1232 Context.get_mechListMIC(),
1233 ),
1234 )
1236 # If we still haven't verified the MIC, we aren't done.
1237 if not Context.verified_mic:
1238 status = GSS_S_CONTINUE_NEEDED
1240 Context.state = SPNEGOSSP.STATE.MICONLY
1242 # Add the output token if provided
1243 if output_token_inner:
1244 spnego_tok.token.responseToken = SPNEGO_Token(
1245 value=output_token_inner,
1246 )
1248 # Update the state
1249 if Context.state != SPNEGOSSP.STATE.MICONLY:
1250 Context.state = SPNEGOSSP.STATE.SUBSEQUENT
1252 return Context, spnego_tok, status
1254 def GSS_Accept_sec_context(
1255 self,
1256 Context: CONTEXT,
1257 input_token=None,
1258 req_flags: Optional[GSS_S_FLAGS] = GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS,
1259 chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,
1260 ):
1261 if Context is None:
1262 # New Context
1263 Context = SPNEGOSSP.CONTEXT(
1264 IsAcceptor=True,
1265 ssps=list(self.ssps),
1266 late_fallback_supported=self.SUPPORT_LATE_FALLBACK,
1267 req_flags=req_flags,
1268 )
1270 input_token_inner = None
1271 _mechListMIC = None
1273 # Extract values from GSSAPI token
1274 if isinstance(input_token, GSSAPI_BLOB):
1275 input_token = input_token.innerToken
1276 if isinstance(input_token, SPNEGO_negToken):
1277 input_token = input_token.token
1278 if isinstance(input_token, SPNEGO_negTokenInit):
1279 # Populate context with values from the client's request
1280 if input_token.mechTypes:
1281 Context.client_mechtypes = input_token.mechTypes
1282 if input_token.mechToken:
1283 input_token_inner = input_token.mechToken.value
1284 _mechListMIC = input_token.mechListMIC or input_token._mechListMIC
1285 elif isinstance(input_token, SPNEGO_negTokenResp):
1286 if input_token.responseToken:
1287 input_token_inner = input_token.responseToken.value
1288 _mechListMIC = input_token.mechListMIC
1289 else:
1290 # The blob is a raw token. We aren't using SPNEGO here.
1291 Context.raw = True
1292 input_token_inner = input_token
1293 self.GuessOtherMechtypes(Context, input_token)
1295 if Context.client_mechtypes is None:
1296 # At this point, we should have already gotten the mechtypes from a current
1297 # or former request.
1298 return Context, None, GSS_S_FAILURE
1300 # Perform SSP negotiation
1301 if Context.ssp is None:
1302 try:
1303 Context.negotiate_ssp()
1304 except ValueError as ex:
1305 # Couldn't find common SSP
1306 log_runtime.warning("SPNEGOSSP: %s" % ex)
1307 return Context, None, GSS_S_FAILURE
1309 output_token_inner = None
1310 status = GSS_S_CONTINUE_NEEDED
1312 # If we didn't pick the client's first choice, the token we were passed
1313 # isn't usable.
1314 if not Context.first_choice:
1315 # Typically a client opportunistically starts with Kerberos, including
1316 # its APREQ, and we want to use NTLM. Here we add one round trip
1317 Context.first_choice = True # Do not enter here again.
1318 else:
1319 # Send it to the negotiated SSP
1320 Context.ssp_context, output_token_inner, status = (
1321 Context.ssp.GSS_Accept_sec_context(
1322 Context.ssp_context,
1323 input_token=input_token_inner,
1324 req_flags=Context.req_flags,
1325 chan_bindings=chan_bindings,
1326 )
1327 )
1329 # Verify MIC if context succeeded
1330 if status == GSS_S_COMPLETE and _mechListMIC:
1331 # NOTE: the mechListMIC that the client sends is computed over the
1332 # **list of mechanisms that it requests**.
1333 if Context.ssp.SupportsMechListMIC():
1334 # We need to check we support checking the MIC. The only case where
1335 # this is needed is NTLM in guest mode: the client will send a mic
1336 # but we don't check it...
1337 Context.ssp.VerifyMechListMIC(
1338 Context.ssp_context,
1339 _mechListMIC.value,
1340 Context.get_mechListMIC(),
1341 )
1342 Context.verified_mic = True
1343 Context.require_mic = True
1345 # Raw processing ends here.
1346 if Context.raw:
1347 return Context, output_token_inner, status
1349 # 0. Build the template response token
1350 spnego_tok = SPNEGO_negToken(
1351 token=SPNEGO_negTokenResp(
1352 supportedMech=None,
1353 )
1354 )
1355 if Context.state == SPNEGOSSP.STATE.FIRST:
1356 # Include the supportedMech list if this is the first message we send
1357 # or a renegotiation.
1358 spnego_tok.token.supportedMech = Context.ssp_mechtype
1360 # Add the output token if provided
1361 if output_token_inner:
1362 spnego_tok.token.responseToken = SPNEGO_Token(value=output_token_inner)
1364 # Update the state
1365 Context.state = SPNEGOSSP.STATE.SUBSEQUENT
1367 # Add the MIC if required and the exchange is finished.
1368 if status == GSS_S_COMPLETE and Context.require_mic:
1369 spnego_tok.token.mechListMIC = SPNEGO_MechListMIC(
1370 value=Context.ssp.GetMechListMIC(
1371 Context.ssp_context,
1372 Context.get_mechListMIC(),
1373 ),
1374 )
1376 # If we still haven't verified the MIC, we aren't done.
1377 if not Context.verified_mic:
1378 status = GSS_S_CONTINUE_NEEDED
1380 # Set negState
1381 spnego_tok.token.negState = self.MapStatusToNegState(status)
1383 return Context, spnego_tok, status
1385 def GSS_Passive(
1386 self,
1387 Context: CONTEXT,
1388 input_token=None,
1389 req_flags=None,
1390 ):
1391 if Context is None:
1392 # New Context
1393 Context = SPNEGOSSP.CONTEXT(
1394 IsAcceptor=True,
1395 ssps=list(self.ssps),
1396 late_fallback_supported=self.SUPPORT_LATE_FALLBACK,
1397 )
1398 Context.passive = True
1400 input_token_inner = None
1402 # Extract values from GSSAPI token
1403 if isinstance(input_token, GSSAPI_BLOB):
1404 input_token = input_token.innerToken
1405 if isinstance(input_token, SPNEGO_negToken):
1406 input_token = input_token.token
1407 if isinstance(input_token, SPNEGO_negTokenInit):
1408 if input_token.mechTypes is not None:
1409 if Context.IsAcceptor:
1410 # NegTokenInit
1411 Context.client_mechtypes = input_token.mechTypes
1412 else:
1413 # NegTokenInit2
1414 Context.server_mechtypes = input_token.mechTypes
1415 if input_token.mechToken:
1416 input_token_inner = input_token.mechToken.value
1417 elif isinstance(input_token, SPNEGO_negTokenResp):
1418 if input_token.supportedMech is not None:
1419 Context.server_mechtypes = [input_token.supportedMech]
1420 if input_token.responseToken:
1421 input_token_inner = input_token.responseToken.value
1422 else:
1423 # Raw.
1424 input_token_inner = input_token
1426 # Get the mechtypes of the other peer
1427 if Context.IsAcceptor:
1428 other_mechtypes = Context.client_mechtypes
1429 else:
1430 other_mechtypes = Context.server_mechtypes
1432 # If we still haven't got a mechtype, guess (raw, most likely)
1433 if other_mechtypes is None:
1434 self.GuessOtherMechtypes(Context, input_token)
1436 # Uninitialized OR allowed mechtypes have changed
1437 if Context.ssp is None or Context.ssp_mechtype not in other_mechtypes:
1438 try:
1439 Context.negotiate_ssp()
1440 except ValueError:
1441 # Couldn't find common SSP
1442 return Context, GSS_S_FAILURE
1444 # Passthrough
1445 Context.ssp_context, status = Context.ssp.GSS_Passive(
1446 Context.ssp_context,
1447 input_token_inner,
1448 req_flags=req_flags,
1449 )
1451 return Context, status
1453 def GSS_Passive_set_Direction(self, Context: CONTEXT, IsAcceptor=False):
1454 Context.IsAcceptor = IsAcceptor
1455 Context.ssp.GSS_Passive_set_Direction(
1456 Context.ssp_context, IsAcceptor=IsAcceptor
1457 )
1459 def MaximumSignatureLength(self, Context: CONTEXT):
1460 return Context.ssp.MaximumSignatureLength(Context.ssp_context)