Improve keytab/ccache support (#4935)

* Improve keytab/ccache support

* Autoargparse: Compat with enums

* LDAPHero: minor improvements

* Ticketer++: better number of tickets

* CLIUtil: improve autocompletion for multi-arg

* Fix PEP8 problems

Author: gpotter2

doc/scapy/layers/kerberos.rst

@@ -238,6 +238,23 @@ As you can see, DMSA keys were imported in the keytab. You can use those as deta
     
     No tickets in CCache.
 
+- **Create server keytab:**
+
+The following is equivalent to Windows' ``ktpass.exe /out kt.keytab /mapuser WKS02$@domain.local /princ host/WKS02.domain.local@domain.local /pass ScapyIsNice``.
+
+.. code:: pycon
+
+    >>> t = Ticketer()
+    >>> t.add_cred("host/WKS02.domain.local@domain.local", etypes="all", mapupn="WKS02$@domain.local", password="ScapyIsNice")
+    Enter password: ************
+    >>> t.show()
+    Keytab name: UNSAVED
+    Principal                              Timestamp          KVNO  Keytype
+    host/WKS02$.domain.local@domain.local  25/02/26 15:40:27  1     AES256-CTS-HMAC-SHA1-96
+
+    No tickets in CCache.
+    >>> t.save_keytab("kt.keytab")
+
 - **Change password using kpasswd in 'set' mode:**
 
 .. code:: pycon
@@ -370,6 +387,10 @@ Cheat sheet
 +---------------------------------------+--------------------------------+
 | ``t.renew(i, [...])``                 | Renew a TGT/ST                 |
 +---------------------------------------+--------------------------------+
+| ``t.remove_krb(i)``                   | Remove a TGT/ST                |
++---------------------------------------+--------------------------------+
+| ``t.set_primary(i)``                  | Set the primary ticket         |
++---------------------------------------+--------------------------------+
 
 Other useful commands
 ---------------------

scapy/compat.py

@@ -3,11 +3,12 @@
 # See https://scapy.net/ for more information
 
 """
-Python 2 and 3 link classes.
+Compatibility module to various older versions of Python
 """
 
 import base64
 import binascii
+import enum
 import struct
 import sys
 
@@ -39,14 +40,15 @@
     'orb',
     'plain_str',
     'raw',
+    'StrEnum',
 ]
 
 # Typing compatibility
 
 # Note:
 # supporting typing on multiple python versions is a nightmare.
 # we provide a FakeType class to be able to use types added on
-# later Python versions (since we run mypy on 3.12), on older
+# later Python versions (since we run mypy on 3.14), on older
 # ones.
 
 
@@ -100,6 +102,15 @@ class Protocol:
 else:
     Self = _FakeType("Self")
 
+
+# Python 3.11 Only
+if sys.version_info >= (3, 11):
+    from enum import StrEnum
+else:
+    class StrEnum(str, enum.Enum):
+        pass
+
+
 ###########
 # Python3 #
 ###########

scapy/layers/kerberos.py

@@ -3033,7 +3033,9 @@ class KerberosClient(Automaton):
     :param dmsa: sets the 'unconditional delegation' mode for DMSA TGT retrieval
     """
 
-    RES_AS_MODE = namedtuple("AS_Result", ["asrep", "sessionkey", "kdcrep", "upn"])
+    RES_AS_MODE = namedtuple(
+        "AS_Result", ["asrep", "sessionkey", "kdcrep", "upn", "pa_type"]
+    )
     RES_TGS_MODE = namedtuple("TGS_Result", ["tgsrep", "sessionkey", "kdcrep", "upn"])
 
     class MODE(IntEnum):
@@ -3232,6 +3234,7 @@ def __init__(
         self.fast_req_sent = False
         # Session parameters
         self.pre_auth = False
+        self.pa_type = None  # preauth-type that's used
         self.fast_rep = None
         self.fast_error = None
         self.fast_skey = None  # The random subkey used for fast
@@ -3574,8 +3577,9 @@ def as_req(self):
                 )
 
                 # Build PA-DATA
+                self.pa_type = 16  # PA-PK-AS-REQ
                 pafactor = PADATA(
-                    padataType=16,  # PA-PK-AS-REQ
+                    padataType=self.pa_type,
                     padataValue=PA_PK_AS_REQ(
                         signedAuthpack=signedAuthpack,
                         trustedCertifiers=None,
@@ -3596,15 +3600,17 @@ def as_req(self):
                         b"clientchallengearmor",
                         b"challengelongterm",
                     )
+                    self.pa_type = 138  # PA-ENCRYPTED-CHALLENGE
                     pafactor = PADATA(
-                        padataType=138,  # PA-ENCRYPTED-CHALLENGE
+                        padataType=self.pa_type,
                         padataValue=EncryptedData(),
                     )
                 else:
                     # Usual 'timestamp' factor
                     ts_key = self.key
+                    self.pa_type = 2  # PA-ENC-TIMESTAMP
                     pafactor = PADATA(
-                        padataType=2,  # PA-ENC-TIMESTAMP
+                        padataType=self.pa_type,
                         padataValue=EncryptedData(),
                     )
                 pafactor.padataValue.encrypt(
@@ -4078,6 +4084,7 @@ def decrypt_as_rep(self, pkt):
             res.key.toKey(),
             res,
             pkt.root.getUPN(),
+            self.pa_type,
         )
 
     @ATMT.receive_condition(SENT_TGS_REQ)

scapy/layers/ldap.py

@@ -26,8 +26,6 @@
 import struct
 import uuid
 
-from enum import Enum
-
 from scapy.arch import get_if_addr
 from scapy.ansmachine import AnsweringMachine
 from scapy.asn1.asn1 import (
@@ -62,6 +60,7 @@
 )
 from scapy.asn1packet import ASN1_Packet
 from scapy.config import conf
+from scapy.compat import StrEnum
 from scapy.error import log_runtime
 from scapy.fields import (
     FieldLenField,
@@ -90,6 +89,7 @@
     GSS_C_FLAGS,
     GSS_C_NO_CHANNEL_BINDINGS,
     GSS_S_COMPLETE,
+    GSS_S_CONTINUE_NEEDED,
     GssChannelBindings,
     SSP,
     _GSSAPI_Field,
@@ -106,6 +106,7 @@
     Any,
     Dict,
     List,
+    Optional,
     Union,
 )
 
@@ -1642,8 +1643,8 @@ def dclocator(
 #####################
 
 
-class LDAP_BIND_MECHS(Enum):
-    NONE = "UNAUTHENTICATED"
+class LDAP_BIND_MECHS(StrEnum):
+    NONE = "ANONYMOUS"
     SIMPLE = "SIMPLE"
     SASL_GSSAPI = "GSSAPI"
     SASL_GSS_SPNEGO = "GSS-SPNEGO"
@@ -1949,8 +1950,8 @@ def bind(
         self,
         mech,
         ssp=None,
-        sign=False,
-        encrypt=False,
+        sign: Optional[bool] = None,
+        encrypt: Optional[bool] = None,
         simple_username=None,
         simple_password=None,
     ):
@@ -1966,6 +1967,12 @@ def bind(
         :
         This acts differently based on the :mech: provided during initialization.
         """
+        # Bind default values: if NTLM then encrypt, else sign unless anonymous/simple
+        if encrypt is None:
+            encrypt = mech == LDAP_BIND_MECHS.SICILY
+        if sign is None and not encrypt:
+            sign = mech not in [LDAP_BIND_MECHS.NONE, LDAP_BIND_MECHS.SIMPLE]
+
         # Store and check consistency
         self.mech = mech
         self.ssp = ssp  # type: SSP
@@ -2000,6 +2007,9 @@ def bind(
         elif mech in [LDAP_BIND_MECHS.NONE, LDAP_BIND_MECHS.SIMPLE]:
             if self.sign or self.encrypt:
                 raise ValueError("Cannot use 'sign' or 'encrypt' with NONE or SIMPLE !")
+        else:
+            raise ValueError("Mech %s is still unimplemented !" % mech)
+
         if self.ssp is not None and mech in [
             LDAP_BIND_MECHS.NONE,
             LDAP_BIND_MECHS.SIMPLE,
@@ -2105,6 +2115,10 @@ def bind(
                 ),
                 chan_bindings=self.chan_bindings,
             )
+            if status not in [GSS_S_COMPLETE, GSS_S_CONTINUE_NEEDED]:
+                raise RuntimeError(
+                    "%s: GSS_Init_sec_context failed !" % self.mech.name,
+                )
             while token:
                 resp = self.sr1(
                     LDAP_BindRequest(
@@ -2116,10 +2130,10 @@ def bind(
                     )
                 )
                 if not isinstance(resp.protocolOp, LDAP_BindResponse):
-                    if self.verb:
-                        print("%s bind failed !" % self.mech.name)
-                        resp.show()
-                    return
+                    raise LDAP_Exception(
+                        "%s bind failed !" % self.mech.name,
+                        resp=resp,
+                    )
                 val = resp.protocolOp.serverSaslCredsData
                 if not val:
                     status = resp.protocolOp.resultCode
@@ -2195,11 +2209,20 @@ def bind(
                     "GSSAPI SASL failed to negotiate client security flags !",
                     resp=resp,
                 )
+
+        # If we use SPNEGO and NTLMSSP was used, understand we can't use sign
+        if self.mech == LDAP_BIND_MECHS.SASL_GSS_SPNEGO:
+            from scapy.layers.ntlm import NTLMSSP
+
+            if isinstance(self.sspcontext.ssp, NTLMSSP):
+                self.sign = False
+
         # SASL wrapping is now available.
         self.sasl_wrap = self.encrypt or self.sign
         if self.sasl_wrap:
             self.sock.closed = True  # prevent closing by marking it as already closed.
             self.sock = StreamSocket(self.sock.ins, LDAP_SASL_Buffer)
+
         # Success.
         if self.verb:
             print("%s bind succeeded !" % self.mech.name)
@@ -2460,3 +2483,4 @@ def close(self):
             print("X Connection closed\n")
         self.sock.close()
         self.bound = False
+        self.sspcontext = None

scapy/layers/smb.py

@@ -286,7 +286,7 @@ class SMBNegotiate_Request(Packet):
 
 bind_layers(SMB_Header, SMBNegotiate_Request, Command=0x72)
 
-# SMBNegociate Protocol Response
+# SMBNegotiate Protocol Response
 
 
 def _SMBStrNullField(name, default):

scapy/layers/smbclient.py

@@ -1323,7 +1323,7 @@ def shares_output(self, results):
         """
         print(pretty_list(results, [("ShareName", "ShareType", "Comment")]))
 
-    @CLIUtil.addcommand(spaces=True)
+    @CLIUtil.addcommand(mono=True)
     def use(self, share):
         """
         Open a share
@@ -1391,7 +1391,7 @@ def _dir_complete(self, arg):
             return [results[0] + "\\"]
         return results
 
-    @CLIUtil.addcommand(spaces=True)
+    @CLIUtil.addcommand(mono=True)
     def ls(self, parent=None):
         """
         List the files in the remote directory
@@ -1466,7 +1466,7 @@ def ls_complete(self, folder):
             return []
         return self._dir_complete(folder)
 
-    @CLIUtil.addcommand(spaces=True)
+    @CLIUtil.addcommand(mono=True)
     def cd(self, folder):
         """
         Change the remote current directory
@@ -1534,7 +1534,7 @@ def lls_output(self, results):
             pretty_list(results, [("FileName", "File Size", "Last Modification Time")])
         )
 
-    @CLIUtil.addcommand(spaces=True)
+    @CLIUtil.addcommand(mono=True)
     def lcd(self, folder):
         """
         Change the local current directory
@@ -1663,7 +1663,7 @@ def _getr(self, directory, _root, _verb=True):
                     print(conf.color_theme.red(remote), "->", str(ex))
         return size
 
-    @CLIUtil.addcommand(spaces=True, globsupport=True)
+    @CLIUtil.addcommand(mono=True, globsupport=True)
     def get(self, file, _dest=None, _verb=True, *, r=False):
         """
         Retrieve a file
@@ -1703,7 +1703,7 @@ def get_complete(self, file):
             return []
         return self._fs_complete(file)
 
-    @CLIUtil.addcommand(spaces=True, globsupport=True)
+    @CLIUtil.addcommand(mono=True, globsupport=True)
     def cat(self, file):
         """
         Print a file
@@ -1731,7 +1731,7 @@ def cat_complete(self, file):
             return []
         return self._fs_complete(file)
 
-    @CLIUtil.addcommand(spaces=True, globsupport=True)
+    @CLIUtil.addcommand(mono=True, globsupport=True)
     def put(self, file):
         """
         Upload a file
@@ -1756,7 +1756,7 @@ def put_complete(self, folder):
         """
         return self._lfs_complete(folder, lambda x: not x.is_dir())
 
-    @CLIUtil.addcommand(spaces=True)
+    @CLIUtil.addcommand(mono=True)
     def rm(self, file):
         """
         Delete a file
@@ -1799,7 +1799,7 @@ def backup(self):
             print("Backup Intent: On")
             self.extra_create_options.append("FILE_OPEN_FOR_BACKUP_INTENT")
 
-    @CLIUtil.addcommand(spaces=True)
+    @CLIUtil.addcommand(mono=True)
     def watch(self, folder):
         """
         Watch file changes in folder (recursively)
@@ -1826,7 +1826,7 @@ def watch(self, folder):
             pass
         print("Cancelled.")
 
-    @CLIUtil.addcommand(spaces=True)
+    @CLIUtil.addcommand(mono=True)
     def getsd(self, file):
         """
         Get the Security Descriptor