add option to statically configure SSP challenge

pyrdp/mitm/RDPMITM.py

@@ -240,7 +240,7 @@ def doClientTls(self):
         ntlmSSPState = NTLMSSPState()
         if self.state.ntlmCapture:
             # We are capturing the NLA NTLMv2 hash
-            self.client.segmentation.addObserver(NLAHandler(self.client.tcp, ntlmSSPState, self.getLog("ntlmssp"), ntlmCapture=True))
+            self.client.segmentation.addObserver(NLAHandler(self.client.tcp, ntlmSSPState, self.getLog("ntlmssp"), ntlmCapture=True, challenge=self.state.config.sspChallenge))
             return
 
         self.client.segmentation.addObserver(NLAHandler(self.server.tcp, ntlmSSPState, self.getLog("ntlmssp")))

pyrdp/mitm/cli.py

@@ -130,6 +130,7 @@ def buildArgParser():
                         action="store_true")
     parser.add_argument("--nla-redirection-host", help="Redirection target ip if NLA is enforced", default=None)
     parser.add_argument("--nla-redirection-port", help="Redirection target port if NLA is enforced", type=int, default=None)
+    parser.add_argument("--ssp-challenge", help="Set challenge for SSP authentictation (e.g. 1122334455667788)", type=str, default=None)
 
     return parser
 
@@ -170,6 +171,10 @@ def configure(cmdline=None) -> MITMConfig:
         sys.stderr.write('Error: please provide both --nla-redirection-host and --nla-redirection-port\n')
         sys.exit(1)
 
+    if args.ssp_challenge is not None and len(args.ssp_challenge) != 16:
+        sys.stderr.write('Error: please provide a challenge with 16 characters (64 bits)\n')
+        sys.exit(1)
+
     if args.target:
         targetHost, targetPort = parseTarget(args.target)
     else:
@@ -201,6 +206,7 @@ def configure(cmdline=None) -> MITMConfig:
     config.useGdi = not args.no_gdi
     config.redirectionHost = args.nla_redirection_host
     config.redirectionPort = args.nla_redirection_port
+    config.sspChallenge = args.ssp_challenge
 
     payload = None
     powershell = None

pyrdp/security/nla.py

@@ -24,7 +24,7 @@ class NLAHandler(SegmentationObserver):
     This also logs the hash of NLA connection attempts.
     """
 
-    def __init__(self, sink: IntermediateLayer, state: NTLMSSPState, log: logging.LoggerAdapter, ntlmCapture: bool = False):
+    def __init__(self, sink: IntermediateLayer, state: NTLMSSPState, log: logging.LoggerAdapter, ntlmCapture: bool = False, challenge: str = None):
         """
         Create a new NLA Handler.
         sink: layer to forward packets to.
@@ -36,13 +36,17 @@ def __init__(self, sink: IntermediateLayer, state: NTLMSSPState, log: logging.Lo
         self.ntlmSSPState = state
         self.ntlmSSPParser = NTLMSSPParser()
         self.ntlmCapture = ntlmCapture
+        self.challenge = challenge
         self.log = log
 
-    def getRandChallenge(self):
+    def getChallenge(self):
         """
-        Generate a random 64-bit challenge
+        Return configured challenge or a random 64-bit challenge
         """
-        challenge = b'%016x' % secrets.randbits(16 * 4)
+        if self.challenge is None:
+            challenge = b'%016x' % secrets.randbits(16 * 4)
+        else:
+            challenge = self.challenge
         return codecs.decode(challenge, 'hex')
 
     def onUnknownHeader(self, header, data: bytes):
@@ -53,16 +57,16 @@ def onUnknownHeader(self, header, data: bytes):
             self.ntlmSSPState.setMessage(message)
 
             if message.messageType == NTLMSSPMessageType.NEGOTIATE_MESSAGE and self.ntlmCapture:
-                randomChallenge = self.getRandChallenge()
+                rawChallenge = self.getChallenge()
                 self.log.debug("NTLMSSP Negotiation")
-                challenge: NTLMSSPChallengePDU = NTLMSSPChallengePDU(randomChallenge)
+                challenge: NTLMSSPChallengePDU = NTLMSSPChallengePDU(rawChallenge)
 
                 # There might be no state if server side connection was shutdown
                 if not self.ntlmSSPState:
                     self.ntlmSSPState = NTLMSSPState()
                 self.ntlmSSPState.setMessage(challenge)
-                self.ntlmSSPState.challenge.serverChallenge = randomChallenge
-                data = self.ntlmSSPParser.writeNTLMSSPChallenge('WINNT', randomChallenge)
+                self.ntlmSSPState.challenge.serverChallenge = rawChallenge
+                data = self.ntlmSSPParser.writeNTLMSSPChallenge('WINNT', rawChallenge)
 
             if message.messageType == NTLMSSPMessageType.AUTHENTICATE_MESSAGE:
                 message: NTLMSSPAuthenticatePDU