Windows/Crypto rework: CertTree, Kerberos PKINIT, Netlogon's Kerberos secure channel, better NTLM variants

.github/workflows/cifuzz.yml

@@ -1,7 +1,7 @@
 name: CIFuzz
 
 on:
-  pull_request:
+  push:
     branches: [master]
 
 permissions:

doc/scapy/layers/kerberos.rst

@@ -41,11 +41,11 @@ This section tries to give many usage examples, but isn't exhaustive. For more d
     >>> t.show()
     Tickets:
     0. Administrator@DOMAIN.LOCAL -> krbtgt/DOMAIN.LOCAL@DOMAIN.LOCAL
-    Start time         End time           Renew until        Auth time        
+    Start time         End time           Renew until        Auth time
     31/08/23 11:38:34  31/08/23 21:38:34  31/08/23 21:38:35  31/08/23 01:38:34
 
     1. Administrator@DOMAIN.LOCAL -> host/dc1.domain.local@DOMAIN.LOCAL
-    Start time         End time           Renew until        Auth time        
+    Start time         End time           Renew until        Auth time
     31/08/23 11:39:07  31/08/23 21:38:34  31/08/23 21:38:35  31/08/23 01:38:34
 
 
@@ -68,12 +68,41 @@ This section tries to give many usage examples, but isn't exhaustive. For more d
     >>> # Using the AES-256-SHA1-96 Kerberos Key
    >>> t.request_tgt("Administrator@domain.local", key=Key(EncryptionType.AES256_CTS_HMAC_SHA1_96, bytes.fromhex("63a2577d8bf6abeba0847cded36b9aed202c23750eb9c56b6155be1cc946bb1d")))
 
+- **Request a TGT using PKINIT**:
+
+.. code:: pycon
+
+    >>> from scapy.libs.rfc3961 import EncryptionType
+    >>> load_module("ticketer")
+    >>> t = Ticketer()
+    >>> # If P12:
+    >>> t.request_tgt("Administrator@DOMAIN.LOCAL", p12="admin.pfx", ca="ca.pem")
+    >>> # One could also have used a different cert and key file:
+    >>> t.request_tgt("Administrator@DOMAIN.LOCAL", x509="admin.cert", x509key="admin.key", ca="ca.pem")
+
+- **Request a user TGT with Kerberos armoring (FAST)**
+
+The ``armor_with`` keyword allows to select a ticket to armor the request with.
+
+.. code:: pycon
+
+    >>> load_module("ticketer")
+    >>> t = Ticketer()
+    >>> t.request_tgt("Machine01$@DOMAIN.LOCAL", key=Key(EncryptionType.RC4_HMAC, bytes.fromhex("2b576acbe6bcfda7294d6bd18041b8fe")))
+    >>> t.show()
+    Tickets:
+    0. Machine01$@DOMAIN.LOCAL -> krbtgt/DOMAIN.LOCAL@DOMAIN.LOCAL
+    Start time         End time           Renew until        Auth time
+    31/08/23 11:38:34  31/08/23 21:38:34  31/08/23 21:38:35  31/08/23 01:38:34
+    >>> t.request_tgt("Administrator@domain.local", armor_with=0)  # Armor with ticket n°0
+
 - **Renew a TGT or ST**:
 
 .. code::
 
     >>> t.renew(0)  # renew TGT
     >>> t.renew(1)  # renew ST. Works only with 'host/' SPNs
+    >>> t.renew(1, armor_with=0)  # renew something with armoring
 
 - **Import tickets from a ccache**:
 

scapy/asn1/mib.py

@@ -284,16 +284,24 @@ def load_mib(filenames):
     "1.3.101.113": "Ed448",
 }
 
+#       pkcs3       #
+
+pkcs3_oids = {
+    "1.2.840.113549.1.3": "pkcs-3",
+    "1.2.840.113549.1.3.1": "dhKeyAgreement",
+}
+
 #       pkcs7       #
 
 pkcs7_oids = {
+    "1.2.840.113549.1.7": "pkcs-7",
     "1.2.840.113549.1.7.2": "id-signedData",
 }
 
 #       pkcs9       #
 
 pkcs9_oids = {
-    "1.2.840.113549.1.9": "pkcs9",
+    "1.2.840.113549.1.9": "pkcs-9",
     "1.2.840.113549.1.9.0": "modules",
     "1.2.840.113549.1.9.1": "emailAddress",
     "1.2.840.113549.1.9.2": "unstructuredName",
@@ -724,6 +732,7 @@ def load_mib(filenames):
     secsig_oids,
     nist_oids,
     thawte_oids,
+    pkcs3_oids,
     pkcs7_oids,
     pkcs9_oids,
     attributeType_oids,

scapy/asn1fields.py

@@ -31,7 +31,6 @@
     BER_tagging_enc,
 )
 from scapy.base_classes import BasePacket
-from scapy.compat import raw
 from scapy.volatile import (
     GeneralizedTime,
     RandChoice,
@@ -599,7 +598,7 @@ def build(self, pkt):
         elif val is None:
             s = b""
         else:
-            s = b"".join(raw(i) for i in val)
+            s = b"".join(bytes(i) for i in val)
         return self.i2m(pkt, s)
 
     def i2repr(self, pkt, x):
@@ -642,6 +641,9 @@ class ASN1F_TIME_TICKS(ASN1F_INTEGER):
 #############################
 
 class ASN1F_optional(ASN1F_element):
+    """
+    ASN.1 field that is optional.
+    """
     def __init__(self, field):
         # type: (ASN1F_field[Any, Any]) -> None
         field.flexible_tag = False
@@ -682,6 +684,20 @@ def i2repr(self, pkt, x):
         return self._field.i2repr(pkt, x)
 
 
+class ASN1F_omit(ASN1F_field[None, None]):
+    """
+    ASN.1 field that is not specified. This is simply omitted on the network.
+    This is different from ASN1F_NULL which has a network representation.
+    """
+    def m2i(self, pkt, s):
+        # type: (ASN1_Packet, bytes) -> Tuple[None, bytes]
+        return None, s
+
+    def i2m(self, pkt, x):
+        # type: (ASN1_Packet, Optional[bytes]) -> bytes
+        return b""
+
+
 _CHOICE_T = Union['ASN1_Packet', Type[ASN1F_field[Any, Any]], 'ASN1F_PACKET']
 
 
@@ -769,7 +785,7 @@ def i2m(self, pkt, x):
         if x is None:
             s = b""
         else:
-            s = raw(x)
+            s = bytes(x)
             if hash(type(x)) in self.pktchoices:
                 imp, exp = self.pktchoices[hash(type(x))]
                 s = BER_tagging_enc(s,
@@ -852,11 +868,11 @@ def i2m(self,
             s = x
         elif isinstance(x, ASN1_Object):
             if x.val:
-                s = raw(x.val)
+                s = bytes(x.val)
             else:
                 s = b""
         else:
-            s = raw(x)
+            s = bytes(x)
             if not hasattr(x, "ASN1_root"):
                 # A normal Packet (!= ASN1)
                 return s
@@ -897,7 +913,7 @@ def __init__(self,
         self.cls = cls
         super(ASN1F_BIT_STRING_ENCAPS, self).__init__(  # type: ignore
             name,
-            default and raw(default),
+            default and bytes(default),
             context=context,
             implicit_tag=implicit_tag,
             explicit_tag=explicit_tag

scapy/layers/dcerpc.py

@@ -451,6 +451,14 @@ class RPC_C_AUTHN_LEVEL(IntEnum):
 DCE_C_AUTHN_LEVEL = RPC_C_AUTHN_LEVEL  # C706 name
 
 
+class RPC_C_IMP_LEVEL(IntEnum):
+    DEFAULT = 0x0
+    ANONYMOUS = 0x1
+    IDENTIFY = 0x2
+    IMPERSONATE = 0x3
+    DELEGATE = 0x4
+
+
 # C706 sect 13.2.6.1
 
 
@@ -2766,9 +2774,9 @@ def __init__(self, *args, **kwargs):
         self.ssp = kwargs.pop("ssp", None)
         self.sspcontext = kwargs.pop("sspcontext", None)
         self.auth_level = kwargs.pop("auth_level", None)
-        self.auth_context_id = kwargs.pop("auth_context_id", 0)
         self.sent_cont_ids = []
         self.cont_id = 0  # Currently selected context
+        self.auth_context_id = 0  # Currently selected authentication context
         self.map_callid_opnum = {}
         self.frags = collections.defaultdict(lambda: b"")
         self.sniffsspcontexts = {}  # Unfinished contexts for passive
@@ -3283,7 +3291,6 @@ def __init__(self, *args, **kwargs):
         self.session = DceRpcSession(
             ssp=kwargs.pop("ssp", None),
             auth_level=kwargs.pop("auth_level", None),
-            auth_context_id=kwargs.pop("auth_context_id", None),
             support_header_signing=kwargs.pop("support_header_signing", True),
         )
         super(DceRpcSocket, self).__init__(*args, **kwargs)

scapy/layers/gssapi.py

@@ -34,7 +34,7 @@
     ASN1_Class_UNIVERSAL,
     ASN1_Codecs,
 )
-from scapy.asn1.ber import BERcodec_SEQUENCE
+from scapy.asn1.ber import BERcodec_SEQUENCE, BER_id_dec
 from scapy.asn1.mib import conf  # loads conf.mib
 from scapy.asn1fields import (
     ASN1F_OID,
@@ -104,19 +104,25 @@ class GSSAPI_BLOB(ASN1_Packet):
     @classmethod
     def dispatch_hook(cls, _pkt=None, *args, **kargs):
         if _pkt and len(_pkt) >= 1:
-            if ord(_pkt[:1]) & 0xA0 >= 0xA0:
+            if _pkt[0] & 0xA0 >= 0xA0:
                 from scapy.layers.spnego import SPNEGO_negToken
 
                 # XXX: sometimes the token is raw, we should look from
                 # the session what to use here. For now: hardcode SPNEGO
                 # (THIS IS A VERY STRONG ASSUMPTION)
                 return SPNEGO_negToken
-            if _pkt[:7] == b"NTLMSSP":
+            elif _pkt[:7] == b"NTLMSSP":
                 from scapy.layers.ntlm import NTLM_Header
 
                 # XXX: if no mechTypes are provided during SPNEGO exchange,
                 # Windows falls back to a plain NTLM_Header.
                 return NTLM_Header.dispatch_hook(_pkt=_pkt, *args, **kargs)
+            elif BER_id_dec(_pkt)[0] & 0x7F > 0x60:
+                from scapy.layers.kerberos import Kerberos
+
+                # XXX: Heuristic to detect raw Kerberos packets, when Windows
+                # fallsback or when the parent data hasn't got any mechtype specified.
+                return Kerberos
         return cls
 
 
@@ -454,7 +460,7 @@ class STATE(IntEnum):
     def GSS_Init_sec_context(
         self,
         Context: CONTEXT,
-        token=None,
+        input_token=None,
         target_name: Optional[str] = None,
         req_flags: Optional[GSS_C_FLAGS] = None,
         chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,
@@ -468,7 +474,7 @@ def GSS_Init_sec_context(
     def GSS_Accept_sec_context(
         self,
         Context: CONTEXT,
-        token=None,
+        input_token=None,
         req_flags: Optional[GSS_S_FLAGS] = GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS,
         chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,
     ):
@@ -477,10 +483,21 @@ def GSS_Accept_sec_context(
         """
         raise NotImplementedError
 
+    @abc.abstractmethod
+    def GSS_Inquire_names_for_mech(self) -> List[str]:
+        """
+        Get the available OIDs for this mech, in order of preference.
+        """
+        raise NotImplementedError
+
     # Passive
 
     @abc.abstractmethod
-    def GSS_Passive(self, Context: CONTEXT, token=None):
+    def GSS_Passive(
+        self,
+        Context: CONTEXT,
+        input_token=None,
+    ):
         """
         GSS_Passive: client/server call for the SSP in passive mode
         """
@@ -591,6 +608,9 @@ def GSS_GetMIC(
         message: bytes,
         qop_req: int = GSS_C_QOP_DEFAULT,
     ):
+        """
+        See GSS_GetMICEx
+        """
         return self.GSS_GetMICEx(
             Context,
             [
@@ -609,7 +629,10 @@ def GSS_VerifyMIC(
         Context: CONTEXT,
         message: bytes,
         signature,
-    ):
+    ) -> None:
+        """
+        See GSS_VerifyMICEx
+        """
         self.GSS_VerifyMICEx(
             Context,
             [
@@ -630,6 +653,9 @@ def GSS_Wrap(
         conf_req_flag: bool,
         qop_req: int = GSS_C_QOP_DEFAULT,
     ):
+        """
+        See GSS_WrapEx
+        """
         _msgs, signature = self.GSS_WrapEx(
             Context,
             [
@@ -647,7 +673,14 @@ def GSS_Wrap(
 
     # sect 2.3.4
 
-    def GSS_Unwrap(self, Context: CONTEXT, signature):
+    def GSS_Unwrap(
+        self,
+        Context: CONTEXT,
+        signature,
+    ):
+        """
+        See GSS_UnwrapEx
+        """
         data = b""
         if signature.payload:
             # signature has a payload that is the data. Let's get that payload
@@ -679,19 +712,19 @@ def NegTokenInit2(self):
         """
         return None, None
 
-    def canMechListMIC(self, Context: CONTEXT):
+    def SupportsMechListMIC(self):
         """
-        Returns whether or not mechListMIC can be computed
+        Returns whether mechListMIC is supported or not
         """
-        return False
+        return True
 
-    def getMechListMIC(self, Context, input):
+    def GetMechListMIC(self, Context, input):
         """
         Compute mechListMIC
         """
-        return bytes(self.GSS_GetMIC(Context, input))
+        return self.GSS_GetMIC(Context, input)
 
-    def verifyMechListMIC(self, Context, otherMIC, input):
+    def VerifyMechListMIC(self, Context, otherMIC, input):
         """
         Verify mechListMIC
         """