Publish NTLMSSP_DOMAIN variant

doc/scapy/layers/smb.rst

@@ -246,6 +246,14 @@ A share is identified by a ``name`` and a ``path`` (+ an optional description ca
         )
     )
 
+**Start a SMB server with NTLM auth in an AD, using machine credentials:**
+
+.. note:: This requires an active account with ``WORKSTATION_TRUST_ACCOUNT`` in its ``userAccountControl``. (otherwise you might get ``STATUS_NO_TRUST_SAM_ACCOUNT``)
+
+.. code:: python
+
+    smbserver(ssp=NTLMSSP_DOMAIN(UPN="Computer1@domain.local", HASHNT=bytes.fromhex("7facdc498ed1680c4fd1448319a8c04f")))
+
 **Start a SMB server with Kerberos auth:**
 
 .. code:: python

scapy/layers/msrpce/msnrpc.py

@@ -563,15 +563,16 @@ class NetlogonClient(DCERPC_Client):
 
         >>> cli = NetlogonClient()
         >>> cli.connect_and_bind("192.168.0.100")
-        >>> cli.establishSecureChannel(
+        >>> cli.establish_secure_channel(
         ...     domainname="DOMAIN", computername="WIN10",
         ...     HashNT=bytes.fromhex("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
         ... )
     """
 
     def __init__(
         self,
-        auth_level=DCE_C_AUTHN_LEVEL.NONE,
+        # Default to PRIVACY: see KB5021130
+        auth_level=DCE_C_AUTHN_LEVEL.PKT_PRIVACY,
         verb=True,
         supportAES=True,
         **kwargs,
@@ -645,7 +646,7 @@ def validate_authenticator(self, auth):
         if tempcred != auth.Credential.data:
             raise ValueError("Server netlogon authenticator is wrong !")
 
-    def establishSecureChannel(
+    def establish_secure_channel(
         self,
         computername: str,
         domainname: str,
@@ -669,6 +670,7 @@ def establishSecureChannel(
         # Flow documented in 3.1.4 Session-Key Negotiation
         # and sect 3.4.5.2 for specific calls
         clientChall = os.urandom(8)
+
         # Step 1: NetrServerReqChallenge
         netr_server_req_chall_response = self.sr1_req(
             NetrServerReqChallenge_Request(
@@ -693,13 +695,15 @@ def establishSecureChannel(
             )
             netr_server_req_chall_response.show()
             raise ValueError
+
         # Calc NegotiateFlags
         NegotiateFlags = FlagValue(
             0x602FFFFF,  # sensible default (Windows)
             names=_negotiateFlags,
         )
         if self.supportAES:
             NegotiateFlags += "AES"
+
         # We are either using NetrServerAuthenticate3 or NetrServerAuthenticateKerberos
         if mode == NETLOGON_SECURE_CHANNEL_METHOD.NetrServerAuthenticate3:
             # We use the legacy NetrServerAuthenticate3 function (NetlogonSSP)
@@ -735,6 +739,7 @@ def establishSecureChannel(
                 NetrServerAuthenticate3_Response not in netr_server_auth3_response
                 or netr_server_auth3_response.status != 0
             ):
+                # An error occurred.
                 NegotiatedFlags = None
                 if NetrServerAuthenticate3_Response in netr_server_auth3_response:
                     NegotiatedFlags = FlagValue(
@@ -748,14 +753,19 @@ def establishSecureChannel(
                                 % (NegotiatedFlags ^ NegotiateFlags)
                             )
                         )
+
+                # Show the error
                 print(
                     conf.color_theme.fail(
                         "! %s"
                         % STATUS_ERREF.get(netr_server_auth3_response.status, "Failure")
                     )
                 )
+
+                # If error is unknown, show the packet entirely
                 if netr_server_auth3_response.status not in STATUS_ERREF:
                     netr_server_auth3_response.show()
+
                 raise ValueError
             # Check Server Credential
             if self.supportAES:
@@ -772,8 +782,10 @@ def establishSecureChannel(
                 ):
                     print(conf.color_theme.fail("! Invalid ServerCredential."))
                     raise ValueError
+
             # SessionKey negotiated !
             self.SessionKey = SessionKey
+
             # Create the NetlogonSSP and assign it to the local client
             self.ssp = self.sock.session.ssp = NetlogonSSP(
                 SessionKey=self.SessionKey,
@@ -785,5 +797,6 @@ def establishSecureChannel(
             NegotiateFlags += "Kerberos"
             # TODO
             raise NotImplementedError
+
         # Finally alter context (to use the SSP)
         self.alter_context()

scapy/layers/ntlm.py

@@ -1180,8 +1180,8 @@ class NTLMSSP(SSP):
 
     Server-only arguments:
 
-        :param DOMAIN_NB_NAME: the domain Netbios name (default: DOMAIN)
-        :param DOMAIN_FQDN: the domain FQDN (default: <domain_nb_name>.local)
+        :param DOMAIN_FQDN: the domain FQDN (default: domain.local)
+        :param DOMAIN_NB_NAME: the domain Netbios name (default: strip DOMAIN_FQDN)
         :param COMPUTER_NB_NAME: the server Netbios name (default: SRV)
         :param COMPUTER_FQDN: the server FQDN
                               (default: <computer_nb_name>.<domain_fqdn>)
@@ -1248,9 +1248,9 @@ def __init__(
         PASSWORD=None,
         USE_MIC=True,
         NTLM_VALUES={},
-        DOMAIN_NB_NAME="DOMAIN",
         DOMAIN_FQDN=None,
-        COMPUTER_NB_NAME="SRV",
+        DOMAIN_NB_NAME=None,
+        COMPUTER_NB_NAME=None,
         COMPUTER_FQDN=None,
         IDENTITIES=None,
         DO_NOT_CHECK_LOGIN=False,
@@ -1263,9 +1263,22 @@ def __init__(
         self.HASHNT = HASHNT
         self.USE_MIC = USE_MIC
         self.NTLM_VALUES = NTLM_VALUES
-        self.DOMAIN_NB_NAME = DOMAIN_NB_NAME
-        self.DOMAIN_FQDN = DOMAIN_FQDN or (self.DOMAIN_NB_NAME.lower() + ".local")
-        self.COMPUTER_NB_NAME = COMPUTER_NB_NAME
+        if UPN is not None:
+            from scapy.layers.kerberos import _parse_upn
+
+            try:
+                user, realm = _parse_upn(UPN)
+                if DOMAIN_FQDN is None:
+                    DOMAIN_FQDN = realm
+                if COMPUTER_NB_NAME is None:
+                    COMPUTER_NB_NAME = user
+            except ValueError:
+                pass
+        self.DOMAIN_FQDN = DOMAIN_FQDN or "domain.local"
+        self.DOMAIN_NB_NAME = (
+            DOMAIN_NB_NAME or self.DOMAIN_FQDN.split(".")[0].upper()[:15]
+        )
+        self.COMPUTER_NB_NAME = COMPUTER_NB_NAME or "SRV"
         self.COMPUTER_FQDN = COMPUTER_FQDN or (
             self.COMPUTER_NB_NAME.lower() + "." + self.DOMAIN_FQDN
         )
@@ -1843,7 +1856,8 @@ def _getSessionBaseKey(self, Context, auth_tok):
             return NTLMv2_ComputeSessionBaseKey(
                 ResponseKeyNT, auth_tok.NtChallengeResponse.NTProofStr
             )
-        log_runtime.debug("NTLMSSP: Bad credentials for %s" % username)
+        elif self.IDENTITIES:
+            log_runtime.debug("NTLMSSP: Bad credentials for %s" % username)
         return None
 
     def _checkLogin(self, Context, auth_tok):
@@ -1872,3 +1886,175 @@ def _checkLogin(self, Context, auth_tok):
             if NTProofStr == auth_tok.NtChallengeResponse.NTProofStr:
                 return True
         return False
+
+
+class NTLMSSP_DOMAIN(NTLMSSP):
+    """
+    A variant of the NTLMSSP to be used in server mode that gets the session
+    keys from the domain using a Netlogon channel.
+
+    This has the same arguments as NTLMSSP, but supports the following in server
+    mode:
+
+    :param UPN: the UPN of the machine account to login for Netlogon.
+    :param HASHNT: the HASHNT of the machine account to use for Netlogon.
+    :param PASSWORD: the PASSWORD of the machine acconut to use for Netlogon.
+    :param DC_IP: (optional) specify the IP of the DC.
+
+    Examples::
+
+        >>> mySSP = NTLMSSP_DOMAIN(
+        ...     UPN="Server1@domain.local",
+        ...     HASHNT=bytes.fromhex("8846f7eaee8fb117ad06bdd830b7586c"),
+        ... )
+    """
+
+    def __init__(self, UPN, *args, timeout=3, ssp=None, **kwargs):
+        from scapy.layers.kerberos import KerberosSSP
+
+        # UPN is mandatory
+        kwargs["UPN"] = UPN
+
+        # Either PASSWORD or HASHNT or ssp
+        if "HASHNT" not in kwargs and "PASSWORD" not in kwargs and ssp is None:
+            raise ValueError(
+                "Must specify either 'HASHNT', 'PASSWORD' or "
+                "provide a ssp=KerberosSSP()"
+            )
+        elif ssp is not None and not isinstance(ssp, KerberosSSP):
+            raise ValueError("'ssp' can only be None or a KerberosSSP !")
+
+        # Call parent
+        super(NTLMSSP_DOMAIN, self).__init__(
+            *args,
+            **kwargs,
+        )
+
+        # Treat specific parameters
+        self.DC_IP = kwargs.pop("DC_IP", None)
+        if self.DC_IP is None:
+            # Get DC_IP from dclocator
+            from scapy.layers.ldap import dclocator
+
+            self.DC_IP = dclocator(
+                self.DOMAIN_FQDN,
+                timeout=timeout,
+                debug=kwargs.get("debug", 0),
+            ).ip
+
+        # If logging in via Kerberos
+        self.ssp = ssp
+
+    def _getSessionBaseKey(self, Context, ntlm):
+        """
+        Return the Session Key by asking the DC.
+        """
+        # No user / no domain: skip.
+        if not ntlm.UserNameLen or not ntlm.DomainNameLen:
+            return super(NTLMSSP_DOMAIN, self)._getSessionBaseKey(Context, ntlm)
+
+        # Import RPC stuff
+        from scapy.layers.dcerpc import NDRUnion
+        from scapy.layers.msrpce.msnrpc import (
+            NetlogonClient,
+            NETLOGON_SECURE_CHANNEL_METHOD,
+        )
+        from scapy.layers.msrpce.raw.ms_nrpc import (
+            NetrLogonSamLogonWithFlags_Request,
+            PNETLOGON_NETWORK_INFO,
+            PNETLOGON_AUTHENTICATOR,
+            NETLOGON_LOGON_IDENTITY_INFO,
+            RPC_UNICODE_STRING,
+            STRING,
+        )
+
+        # Create NetlogonClient with PRIVACY
+        client = NetlogonClient()
+        client.connect_and_bind(self.DC_IP)
+
+        # Establish the Netlogon secure channel (this will bind)
+        try:
+            if self.ssp is None:
+                # Login via classic NetlogonSSP
+                client.establish_secure_channel(
+                    mode=NETLOGON_SECURE_CHANNEL_METHOD.NetrServerAuthenticate3,
+                    computername=self.COMPUTER_NB_NAME,
+                    domainname=self.DOMAIN_NB_NAME,
+                    HashNt=self.HASHNT,
+                )
+            else:
+                # Login via KerberosSSP (Windows 2025)
+                # TODO
+                client.establish_secure_channel(
+                    mode=NETLOGON_SECURE_CHANNEL_METHOD.NetrServerAuthenticateKerberos,
+                )
+        except ValueError:
+            log_runtime.warning(
+                "Couldn't establish the Netlogon secure channel. "
+                "Check the credentials for '%s' !" % self.COMPUTER_NB_NAME
+            )
+            return super(NTLMSSP_DOMAIN, self)._getSessionBaseKey(Context, ntlm)
+
+        # Request validation of the NTLM request
+        req = NetrLogonSamLogonWithFlags_Request(
+            LogonServer="",
+            ComputerName=self.COMPUTER_NB_NAME,
+            Authenticator=client.create_authenticator(),
+            ReturnAuthenticator=PNETLOGON_AUTHENTICATOR(),
+            LogonLevel=6,  # NetlogonNetworkTransitiveInformation
+            LogonInformation=NDRUnion(
+                tag=6,
+                value=PNETLOGON_NETWORK_INFO(
+                    Identity=NETLOGON_LOGON_IDENTITY_INFO(
+                        LogonDomainName=RPC_UNICODE_STRING(
+                            Buffer=ntlm.DomainName,
+                        ),
+                        ParameterControl=0x00002AE0,
+                        UserName=RPC_UNICODE_STRING(
+                            Buffer=ntlm.UserName,
+                        ),
+                        Workstation=RPC_UNICODE_STRING(
+                            Buffer=ntlm.Workstation,
+                        ),
+                    ),
+                    LmChallenge=Context.chall_tok.ServerChallenge,
+                    NtChallengeResponse=STRING(
+                        Buffer=bytes(ntlm.NtChallengeResponse),
+                    ),
+                    LmChallengeResponse=STRING(
+                        Buffer=bytes(ntlm.LmChallengeResponse),
+                    ),
+                ),
+            ),
+            ValidationLevel=6,
+            ExtraFlags=0,
+            ndr64=client.ndr64,
+        )
+
+        # Get response
+        resp = client.sr1_req(req)
+        if resp and resp.status == 0:
+            # Success
+
+            # Validate DC authenticator
+            client.validate_authenticator(resp.ReturnAuthenticator.value)
+
+            # Get and return the SessionKey
+            UserSessionKey = resp.ValidationInformation.value.value.UserSessionKey
+            return bytes(UserSessionKey)
+        else:
+            # Failed
+            from scapy.layers.smb2 import STATUS_ERREF
+
+            print(
+                conf.color_theme.fail(
+                    "! %s" % STATUS_ERREF.get(resp.status, "Failure !")
+                )
+            )
+            if resp.status not in STATUS_ERREF:
+                resp.show()
+            return super(NTLMSSP_DOMAIN, self)._getSessionBaseKey(Context, ntlm)
+
+    def _checkLogin(self, Context, auth_tok):
+        # Always OK if we got the session key
+        return True

scapy/layers/smbclient.py

@@ -1227,6 +1227,7 @@ def __init__(
             ssp=ssp,
             debug=debug,
             REQUIRE_ENCRYPTION=REQUIRE_ENCRYPTION,
+            timeout=timeout,
             **kwargs,
         )
         try: